package com.myNetty;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class MyEventLoop implements EventLoop{

    private static final AtomicInteger THREAD_NAME_INDEX = new AtomicInteger();

    private final BlockingQueue<Runnable> taskQueue;

    private final PriorityBlockingQueue<scheduleTask> scheduleTasks;

    private final Thread thread;

    private static final Runnable WAKE_UP = ()->{};

    public MyEventLoop(){
        thread = new EventLoopThread("my-eventloop-thread-" + THREAD_NAME_INDEX.incrementAndGet());
        taskQueue = new ArrayBlockingQueue<>(1024);
        scheduleTasks = new PriorityBlockingQueue<>(1024);
        thread.start();
    }

    @Override
    public void execute(Runnable runnable) {
        if (!taskQueue.offer(runnable)) {
            throw new RuntimeException("TaskQueue is full");
        }
    }

    @Override
    public void schedule(Runnable task, long delay, TimeUnit unit) {
        scheduleTask scheduleTask = new scheduleTask(task, deadlineMs(delay, unit), -1, this);
        if (!scheduleTasks.offer(scheduleTask)) {
            throw new RuntimeException("ScheduleTaskQueue is full");
        }
        execute(WAKE_UP);
    }

    @Override
    public void scheduleAtFixedRate(Runnable task, long initialDelay, long period, TimeUnit unit) {
        scheduleTask scheduleTask = new scheduleTask(task, deadlineMs(initialDelay, unit), unit.toMillis(period), this);
        if (!scheduleTasks.offer(scheduleTask)) {
            throw new RuntimeException("ScheduleTaskQueue is full");
        }
        execute(WAKE_UP);
    }

    @Override
    public EventLoop next() {
        return null;
    }

    private long deadlineMs(long delay,TimeUnit unit){
        return unit.toMillis(delay) + System.currentTimeMillis();
    }

    private Runnable getTask(){
        scheduleTask peek = scheduleTasks.peek();
        if(peek == null){
            Runnable task = null;
            try{
                // take 会阻塞队列，时间无限
                // 加入定时任务时，这里需要唤醒，进行刷新。
                task = taskQueue.take();
                if(task == WAKE_UP){
                    task = null;
                }
            }catch (InterruptedException ignore){

            }
            return task;
        }
        // 当前时间达到预定时间时，执行任务
        if(peek.getDeadline() <= System.currentTimeMillis()){
            // poll 不会阻塞
            return scheduleTasks.poll();
        }
        Runnable task = null;
        try {
            // 这里poll函数作用是，超时阻塞。
            // 即阻塞指定时间，若期间能获取到则返回。超时仍然未获取，则返回空。
            // 这样就是在等待定时任务期间，对于突然加入的普通任务进行进行执行
            // 但是在对于加入的新的定时任务可能会出现问题，所以需要wakeup进行一次唤醒刷新。
            task = taskQueue.poll(peek.getDeadline() - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
            if(task == WAKE_UP){
                task = null;
            }
        }catch (InterruptedException ignore){

        }
        return task;
    }

    public PriorityBlockingQueue<scheduleTask> getScheduleTasks() {
        return scheduleTasks;
    }

    class EventLoopThread extends Thread{

        public EventLoopThread(String name){
            super(name);
        }

        @Override
        public void run() {
            while (true){
                Runnable task = getTask();
                task.run();
            }
        }
    }
}
